package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.exception.AuthorizationException;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.storage.transaction.TransactionLog;
import com.aiwiown.snackmq.storage.transaction.TransactionRecord;
import com.aiwiown.snackmq.storage.transaction.TransactionState;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.UUID;

/**
 * 处理事务消息的第一阶段：接收并存储半消息。
 */
@Slf4j
@RequiredArgsConstructor
public class PrepareMessageHandler implements MessageHandler {

    private final TransactionLog transactionLog;
    private final AuthorizationService authorizationService;

    @Override
    public MessageType getMessageType() {
        return MessageType.PREPARE_TRANSACTIONAL_MESSAGE;
    }

    @Override
    public void handle(RequestContext ctx, Message request) {
        try {
            // 1. 授权检查：确保用户有权向目标主题生产消息
            authorizationService.authorize(
                    request.getProperty("authToken"),
                    Action.PRODUCE,
                    new AuthorizationService.Resource("TOPIC", request.getTopic())
            );

            // 2. 生成全局唯一的事务ID
            String transactionId = UUID.randomUUID().toString();
            // --- 【新增】从请求中获取 producerId ---
            String producerId = request.getProperty("producerId");
            if (producerId == null || producerId.isEmpty()) {
                ctx.writeResponse(Message.builder()
                        .messageId(request.getMessageId())
                        .type(MessageType.RESPONSE)
                        .status(MessageStatus.FAILED)
                        .errorMessage("producerId is missing in PREPARE_TRANSACTIONAL_MESSAGE")
                        .build());
                return;
            }
            // 3. 将事务ID附加到消息中
            Message messageWithTxId = request.toBuilder()
                    .property("transactionId", transactionId)
                    .build();

            // 4. 创建 PREPARE 状态的事务记录，并包含 producerId
            TransactionRecord record = new TransactionRecord(
                    transactionId,
                    TransactionState.PREPARE,
                    System.currentTimeMillis(),
                    messageWithTxId,
                    producerId // 【修改】保存 producerId
            );
            // 5. 将记录写入事务日志
            transactionLog.append(record);

            // 6. 向生产者返回成功响应，并附带事务ID
            Message response = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.SUCCESS)
                    .property("transactionId", transactionId)
                    .build();
            ctx.writeResponse(response);

            log.debug("Successfully prepared transaction {}. Message: {}", transactionId, request.getMessageId());

        } catch (AuthorizationException e) {
            log.warn("Authorization failed for prepare transaction request: {}", e.getMessage());
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Permission denied: " + e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        } catch (IOException e) {
            log.error("Failed to write to transaction log for message: {}", request.getMessageId(), e);
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Failed to persist prepare message due to storage error.")
                    .build();
            ctx.writeResponse(errorResponse);
        }
    }
}